package datastore.api.impl;

import java.util.Iterator;

import org.apache.hadoop.hbase.client.Scanner;
import org.apache.hadoop.hbase.io.RowResult;

public class ScannerIterable<M> implements Iterable<M> {

    protected Scanner scanner;
    protected RowResultHandler rowResultHandler;
    protected int limit;
    protected int count;

    public ScannerIterable(Scanner scanner, int limit, int offset, RowResultHandler rowResultHandler) throws Exception {
        super();
        this.scanner = scanner;
        this.limit = limit;
        this.count = 0;
        this.rowResultHandler = rowResultHandler;
        if (offset > 0) {
            this.scanner.next(offset);
        }
    }

    @Override
    public Iterator<M> iterator() {
        return new ScannerIterator();
    }

    public class ScannerIterator implements Iterator<M> {
        
        private Iterator<RowResult> iterator;

        public ScannerIterator() {
            this.iterator = scanner.iterator();
        }

        @Override
        public boolean hasNext() {
            return count < limit && iterator.hasNext();
        }

        @Override
        @SuppressWarnings("unchecked")
        public M next() {
            try {
                RowResult next = iterator.next();
                count++;
                return (M) rowResultHandler.get(next);
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }

        @Override
        public void remove() {
            iterator.remove();
        }

    }

    public static abstract class RowResultHandler {

        public abstract Object get(RowResult rowResult) throws Exception;

    }


}
